package mapreduce;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.lang.reflect.Constructor;
import java.util.concurrent.CountDownLatch;

/**
 * The Map Task
 */
public class Maptask<KEYIN, VALUEIN, KEYOUT, VALUEOUT> extends Task {

	/*
	 * (non-javadoc)
	 */
	private MapTaskConf<KEYIN, VALUEIN, KEYOUT, VALUEOUT> mapTaskConf = null;
	private Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> mapper;
	private CountDownLatch end;

	public Maptask(CountDownLatch end, int taskid,
			MapTaskConf<KEYIN, VALUEIN, KEYOUT, VALUEOUT> conf) {
		super(taskid);
		this.mapTaskConf = conf;
		this.end = end;
	}

	@SuppressWarnings("unchecked")
	@Override
	public void run() {
		try {
			String line;
			String[] lines;
			File file = new File(mapTaskConf.getFileAddress());
			Class<?> mapclass = mapTaskConf.getMapClass();
			Constructor<?> constructor = mapclass.getConstructor();
			mapper = (Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>) constructor
					.newInstance();
			BufferedReader br = new BufferedReader(new FileReader(file));
			Object key = null, value = null;
			Class<?> inputKeyClass = mapTaskConf.getInputKeyClass();
			Class<?> inputValueClass = mapTaskConf.getInputValueClass();

			while ((line = br.readLine()) != null) {
				lines = line.split("\t");
				if (inputKeyClass == Integer.class)
					key = Integer.parseInt(lines[0]);
				else if (inputKeyClass == Double.class)
					key = Double.parseDouble(lines[0]);
				else
					key = lines[0];
				if (inputValueClass == Integer.class)
					value = Integer.parseInt(lines[1]);
				else if (inputValueClass == Double.class)
					value = Double.parseDouble(lines[1]);
				else
					value = lines[1];

				mapper.map((KEYIN) key, (VALUEIN) value, mapTaskConf);
			}

			mapTaskConf.close();
			end.countDown();

		} catch (Exception e) {
			// TODO: handle exception
		}

	}

}
